package com.ibm.cps.api;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import javax.inject.Named;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriInfo;

import com.ibm.cps.kafka.KafkaTopicConstructorForMultiTenants;
import com.ibm.cps.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.jarxs.ProcessorResource;
import com.ibm.cps.kafka.KafkaProducer;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.MetadataTopologySort;
import com.ibm.cps.processors.ProcessorsGroup;
import com.ibm.cps.servlet.AssignIdFactory;
import com.ibm.cps.servlet.ProcessorsGetFactory;
import com.ibm.factories.PublishProcessorFactory;
import com.ibm.interfaces.IPersistency;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

@Named
public class ProcessorBean implements ProcessorResource{

	private Logger logger;
	private KafkaProducer producer;
	private IPersistency persistency;
	private ProcessorMetadtaValidator processorValidator;
	private AuthorizationManager authManager;
	private KafkaTopicConstructorForMultiTenants kafkaTopicConstructorForMultiTenants = new KafkaTopicConstructorForMultiTenants();

	public ProcessorBean() throws CPSException {
		logger = LoggerFactory.getLogger(this.getClass());
		producer = new KafkaProducer(LocalConfig.KAFKA_BROKERS);
		logger.info("Connect to kafka " + LocalConfig.KAFKA_BROKERS);
		persistency = MongoPersistency.getInstace();
		authManager = AuthorizationManager.getInstance();
		
		logger.info("Connect to persistency " + LocalConfig.MONGO_CONNECT_STRING + ", database is "
				+ LocalConfig.PERSISTENCY_DATABASE);
		processorValidator = ProcessorMetadtaValidator.getInstance();
	}
	



	@Override
	public Response GetProcessors(String tenantid, UriInfo info,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
		ObjectNode filter = JSONUtility.newObjectNode(MessageFields.TENANTID, tenantid);
		MultivaluedMap<String, String>  map = info.getQueryParameters();
		for (String field: map.keySet()) {
			if (field.equals("q") || field.equals("source")) {
				continue;
			}
			filter.put(field, map.getFirst(field));
		}
		
		String str;
		try {
			String source = info.getQueryParameters().getFirst(MessageFields.SOURCE);
			source = (source == null )? "rest":source;
			DataSourceMonitor monitor = DataSourceMonitor.getInstance();
			PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
			MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
//			Collection<String> metadatas = persistency.getProcessors(tenantid, null,null, null);
			Collection<String> metadatas = persistency.getProcessors(filter);
			// Collection<String> metadatas =
			// getMetadatas(key2Value.get(MessageFields.SOURCE), tenantid,
			// key2Value.get(MessageFields.TYPE),
			// key2Value.get(MessageFields.GROUPID),
			// key2Value.get(MessageFields.PROCESSORID));
			Collection<String> allMetadata = persistency.getProcessors(tenantid, null, null, null);
			// Collection<String> allMetadata =
			// persistency.getMetadataByUser(tenantid);
			Collection<String> result = adapter.getJsons(allMetadata, metadatas);
			str = ServletUtil.constructCollectionJson(result);
			return Response.ok(str).type(MediaType.APPLICATION_JSON).build();
		} catch (CPSException e) {
			e.printStackTrace();
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}

	
	}

	@Override
	public Response GetProcessor(String tenantid,String groupid, UriInfo info,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
		ObjectNode filter = JSONUtility.newObjectNode(MessageFields.TENANTID, tenantid);
		filter.put(MessageFields.GROUPID, groupid);
		MultivaluedMap<String, String>  map = info.getQueryParameters();
		for (String field: map.keySet()) {
			if (field.equals("q") || field.equals("source")) {
				continue;
			}
			filter.put(field, map.getFirst(field));
		}
		
		String str;
		try {
			String source = info.getQueryParameters().getFirst(MessageFields.SOURCE);
			source = (source == null )? "rest":source;
			DataSourceMonitor monitor = DataSourceMonitor.getInstance();
			PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
			MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
//			Collection<String> metadatas = persistency.getProcessors(tenantid, null,null, null);
			Collection<String> metadatas = persistency.getProcessors(filter);
			// Collection<String> metadatas =
			// getMetadatas(key2Value.get(MessageFields.SOURCE), tenantid,
			// key2Value.get(MessageFields.TYPE),
			// key2Value.get(MessageFields.GROUPID),
			// key2Value.get(MessageFields.PROCESSORID));
			Collection<String> allMetadata = persistency.getProcessors(tenantid, null, null, null);
			// Collection<String> allMetadata =
			// persistency.getMetadataByUser(tenantid);
			Collection<String> result = adapter.getJsons(allMetadata, metadatas);
			str = ServletUtil.constructCollectionJson(result);
			return Response.ok(str).build();
		} catch (CPSException e) {
			e.printStackTrace();
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}

	
	}

	@Override
	public Response PutProcessors(String tenantid, String groupid, UriInfo info,String json,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
		ObjectNode node = JSONUtility.newObjectNode("message", "need to implement");
		node.put("Code", 400);
		return Response.ok(node.toString(), MediaType.APPLICATION_JSON).build();
	}

	@Override
	public Response PostProcessors(String tenantid,UriInfo info,String json,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
//		if (info.getQueryParameters().getFirst(MessageFields.SOURCE) == null) {
//			ObjectNode node = JSONUtility.newObjectNode("message", "need to specific a source:[node-red/sql]");
//			node.put("code", 400);
//			return Response.ok(node.toString(), MediaType.APPLICATION_JSON).build();
//		}
		
		ArrayNode retArray = JSONUtility.newArrayNode();
		ArrayList<ObjectNode> sendMsg = new ArrayList<ObjectNode>();
		ArrayList<ObjectNode> groups = new ArrayList<ObjectNode>();
		String source = info.getQueryParameters().getFirst(MessageFields.SOURCE);
		if (source == null) {
			source = "rest";
		}
		
		
		try {
			
			DataSourceMonitor monitor = DataSourceMonitor.getInstance();
			PublishProcessorFactory publishFactory = PublisherMonitor.getInstance().getFactory(tenantid);
			MetadataAdapter adapter = AdapaterFactory.getAdapter(monitor.getFactory(tenantid), publishFactory, source);
			json = adapter.adapt2Json(json, source, tenantid);
			if (json == null) {
				String result = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Null input metadata.");
				return Response.ok(result, MediaType.APPLICATION_JSON).build();
			}

			logger.info("Adapter output: " + json);
			ArrayNode objects = JSONUtility.getArrayFromString(json);

			Collection<String> existedMetadatas = persistency.getProcessors(tenantid, null, null, null);
			ArrayNode arrayNode = JSONUtility.newArrayNode(existedMetadatas);
			if (arrayNode != null) {
				arrayNode.addAll(objects);
				MetadataTopologySort.getSortedMetadatas(arrayNode);
			}
			List<ObjectNode> metadatas = MetadataTopologySort.getSortedMetadatas(objects);
			for (ObjectNode object : metadatas) {
				String type = JSONUtility.GetString(MessageFields.TYPE, object);
				String result = validateMetadata(object, type, source, tenantid);
				if (!ProcessorsGroup.isGroupType(type)) {
					sendMsg.add(object);
				} else {
					groups.add(object);
				}
				if (result == null) {
					continue;
				}
				ObjectNode node = JSONUtility.fromObjectString(result);
				retArray.add(node);
			}

			// if all metadata pass the validation, then send all of them to
			// persistency and kafka
			for (ObjectNode metadata : sendMsg) {
				producer.send(kafkaTopicConstructorForMultiTenants.getMetadataAddTopic(tenantid), metadata.toString().getBytes());
				persistency.putProcessor(metadata);
				logger.info("Put message into persistency:" + metadata);
			}

			for (ObjectNode groupMetadata : groups) {
				persistency.putGroupMetadata(groupMetadata);
				// persistency.putMetadata(tenantid, groupMetadata.toString());
			}

			ObjectNode object =  JSONUtility.newObjectNode();
			object.put("Code", 201);
			return Response.ok(object.toString(),MediaType.APPLICATION_JSON).build();
		} catch (CPSException e) {
			e.printStackTrace();
			try {
				for (ObjectNode tmpObject : sendMsg) {
					processorValidator.deleteProcessor(tenantid, tmpObject.toString(), source);
				}
				for (ObjectNode tmpObject : groups) {
					processorValidator.deleteProcessor(tenantid, tmpObject.toString(), source);
				}
			} catch (CPSException e1) {
				e1.printStackTrace();
			}
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}

	
		
	}
	
	@Override
	public Response DeleteProcessors(String tenantid,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
		String source = "rest";
		try {
			Collection<String> deletedMetadata = ProcessorsGetFactory.getMetadatasAndGroups(source,
					tenantid,null, null,null);

			String result;
			if (deletedMetadata != null) {
				 result = deleteProcessors(tenantid, deletedMetadata, source);
			} else {
				result = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
			}
			
			return Response.ok(result).type(MediaType.APPLICATION_JSON).build();
			
		} catch (CPSException e) {
			e.printStackTrace();
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}
	

	
	}

	@Override
	public Response DeleteProcessor(String tenantid, String groupid,String key) {
		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
		String source = "rest";
		try {
			Collection<String> deletedMetadata = ProcessorsGetFactory.getMetadatasAndGroups(source,
					tenantid, null, groupid , null);

			String result;
			if (deletedMetadata != null) {
				 result = deleteProcessors(tenantid, deletedMetadata, source);
			} else {
				result = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
			}
			
			return Response.ok(result).build();
			
		} catch (CPSException e) {
			e.printStackTrace();
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}
	

	
	}
	
	
	@Override
	public Response DeleteProcessor(String tenantid, String groupid, String processor, String key) {

		if (!authManager.isAllow(tenantid, key)) {
			String ret = ServletUtil.constructReturnMsg(400, "invalid access");
			return Response.ok(ret,MediaType.APPLICATION_JSON).build();
		}
		
		String source = "rest";
		try {
			Collection<String> deletedMetadata = ProcessorsGetFactory.getMetadatasAndGroups(source,
					tenantid, null, groupid , processor);

			String result;
			if (deletedMetadata != null) {
				 result = deleteProcessors(tenantid, deletedMetadata, source);
			} else {
				result = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
			}
			
			return Response.ok(result).build();
			
		} catch (CPSException e) {
			e.printStackTrace();
			String ret = ServletUtil.constructReturnMsg(e.getErrorCode(), e.getErrorMessage());
			return Response.ok(ret).type(MediaType.APPLICATION_JSON).build();
		}

	}
	
	
	/**
	 * validate processor/group metadata
	 * 
	 * @param object
	 * @param type
	 * @param source
	 * @param tenantid
	 * @return
	 * @throws CPSException
	 */
	private String validateMetadata(ObjectNode object, String type, String source, String tenantid)
			throws CPSException {
		
		String processorid = AssignIdFactory.getProcessorid(type, object);
		boolean success = processorValidator.validate(tenantid, processorid, object, source);
		if (!success) {
			return null;
		}
		return constructRetMsg(source, processorid, type);
	}
	
	/**
	 * construct return message according to different conditions
	 * 
	 * @param source
	 * @param processorid
	 * @param type
	 * @return
	 */
	private String constructRetMsg(String source, String processorid, String type) {
		if (ProcessorsGroup.isGroupType(type)) {
			ObjectNode retNode = JSONUtility.newObjectNode();
			retNode.put("id", processorid);
			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, retNode);
		}
		if (source.equals(MessageFields.GROUP_TYPE_SQL) || source.equals(MessageFields.NODERED)) {
			return null;
		} else {
			return ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, processorid);
		}
	}
	
	
	private String deleteProcessors(String tenantid, Collection<String> deletedMetadata, String source)
			throws CPSException {
		String retMessage;
		ArrayNode deletedIds = JSONUtility.newArrayNode();
		if (deletedMetadata != null && deletedMetadata.size() >= 1) {
			for (String deleted : deletedMetadata) {
				ArrayNode tmp = processorValidator.deleteProcessor(tenantid, deleted, source);
				if (tmp != null && tmp.size() >= 1) {
					for (int index = 0; index < tmp.size(); index++) {
						deletedIds.add(tmp.get(index));
					}
					producer.send(kafkaTopicConstructorForMultiTenants.getMetadataDeleteTopic(tenantid), deleted.getBytes());
				}
			}
			if (deletedIds.size() >= 1) {
				retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE,
						deletedIds.toString() + " have been deleted.");
			} else {
				retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing has been deleted.");
			}
		} else {
			retMessage = ServletUtil.constructReturnMsg(ErrorCode.SUCCESS_CODE, "Nothing to be deleted.");
		}
		return retMessage;
	}

}
